package com.at.bigdata.spark.core.acc

import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

/**
 * 依赖关系
 *
 * @author cdhuangchao3
 * @date 2023/3/6 8:40 PM
 */
object Spark06_Bc {

  def main(args: Array[String]): Unit = {
    val sparConf = new SparkConf().setMaster("local").setAppName("WordCount")
    val sc = new SparkContext(sparConf)

    val rdd1 = sc.makeRDD(List(
      ("a", 1), ("b", 2), ("c", 3)
    ))

    //    val rdd2 = sc.makeRDD(List(
    //      ("a", 4), ("b", 5), ("c", 6)
    //    ))
    // 闭包数据，都是以Task为单位发送的，每个任务重包含闭包数据
    // 这样可能会导致，一个Executor中含有大量重复的数据，并且占用大量的内存
    val map = mutable.Map(("a", 4), ("b", 5), ("c", 6))
    // 封装广播变量
    val bc = sc.broadcast(map)
    // Executor其实就一个JVM，所以在启动时，会自动分配内存
    // 完全可以将任务重的闭包数据，放置是Executor的内存中，达到共享的目的
    // Spark中的广播变量就可以将闭包的数据保存到Executor的内存中
    // Spark中的广播变量不能更改：分布式共享只读变量

    // join会导致数据量的几何增长，并且会影响shuffle的性能，不推荐使用
    //    val joinRDD = rdd1.join(rdd2)

    //    joinRDD.collect().foreach(println)
    // (a, (1,4)), (b, (2,5)), (c, (3,6))

    rdd1.map {
      case (w, c) => {
        val l = bc.value.getOrElse(w, 0)
        (w, (c, l))
      }
    }.collect().foreach(println)

    sc.stop()
  }
}
